package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TradeProvinceOrderBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.function.AsyncDimFunction;
import com.atguigu.gmall.realtime.function.DorisMapFunction;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/8/14 08:30
 */
public class Dws_10_DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow().start(
            40010,
            2,
            "Dws_10_DwsTradeProvinceOrderWindow",
            GmallConstant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TradeProvinceOrderBean> beanStreamWithOoutDims = stream
            .map(new MapFunction<String, TradeProvinceOrderBean>() {
                @Override
                public TradeProvinceOrderBean map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    return TradeProvinceOrderBean.builder()
                        .provinceId(obj.getString("province_id"))
                        .orderId(obj.getString("order_id"))
                        .orderDetailId(obj.getString("id"))
                        .orderAmount(obj.getBigDecimal("split_total_amount"))
                        .ts(obj.getLong("ts") * 1000)
                        .build();
                }
            })
            .keyBy(TradeProvinceOrderBean::getOrderDetailId)  // 先按照详情 id 去重,再按照 orderId 找到这个订单的第一条详情
            .process(new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                private ValueState<TradeProvinceOrderBean> lastBeanState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<TradeProvinceOrderBean> desc = new ValueStateDescriptor<>("lastBean", TradeProvinceOrderBean.class);
                    StateTtlConfig ttlConf = StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(50000))
                        .updateTtlOnCreateAndWrite()
                        .neverReturnExpired()
                        .build();
                    desc.enableTimeToLive(ttlConf);
                    lastBeanState = getRuntimeContext().getState(desc);
                }
                
                @Override
                public void processElement(TradeProvinceOrderBean bean,
                                           Context ctx,
                                           Collector<TradeProvinceOrderBean> out) throws Exception {
                    TradeProvinceOrderBean lastBean = lastBeanState.value();
                    
                    if (lastBean == null) { // 第一条
                        // 当前数据直接输出
                        out.collect(bean);
                    } else {  // 不是第一条
                        // 上一条数据取反 + 当前数据, 然后输出 或者 当前 - 上一条
                        lastBean.setOrderAmount(bean.getOrderAmount().subtract(lastBean.getOrderAmount()));
                        out.collect(lastBean);
                    }
                    
                    // 把当前数据存入到状态
                    lastBeanState.update(bean);
                    
                    
                }
            })
            .keyBy(TradeProvinceOrderBean::getOrderId)  // 按照订单id 分组,找到这个订单的第一个详情,对订单数贡献 1, 这个订单的其他详情对订单数贡献 0
            .process(new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                
                private ValueState<Boolean> isFirstDetailState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    isFirstDetailState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstDetail", Boolean.class));
                }
                
                @Override
                public void processElement(TradeProvinceOrderBean value,
                                           Context ctx,
                                           Collector<TradeProvinceOrderBean> out) throws Exception {
                    Boolean isFirstDetail = isFirstDetailState.value();
                    value.setOrderCount(0L);
                    if (isFirstDetail == null) {
                        isFirstDetailState.update(false);
                        value.setOrderCount(1L);
                    }
                    out.collect(value);
                    
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeProvinceOrderBean::getProvinceId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1,
                                                         TradeProvinceOrderBean value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String provinceId,
                                        Context ctx,
                                        Iterable<TradeProvinceOrderBean> elements,
                                        Collector<TradeProvinceOrderBean> out) throws Exception {
                        
                        TradeProvinceOrderBean bean = elements.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(ctx.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(ctx.window().getStart()));
                        
                        out.collect(bean);
                        
                    }
                }
            );
    
        AsyncDataStream.unorderedWait(
                beanStreamWithOoutDims,
                new AsyncDimFunction<TradeProvinceOrderBean>() {
                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }
                
                    @Override
                    public String getId(TradeProvinceOrderBean bean) {
                        return bean.getProvinceId();
                    }
                
                    @Override
                    public void addDim(TradeProvinceOrderBean bean, JSONObject dim) {
                        bean.setProvinceName(dim.getString("name"));
                    }
                },
                120,
                TimeUnit.SECONDS
            )
            .map(new DorisMapFunction<>())
            .sinkTo(DorisUtil.getDorisSink("gmall2023.dws_trade_province_order_window"));
        
        
    }
}
/*
从 Kafka 读取订单明细数据，过滤 null 数据并按照唯一键对数据去重，
统计各省份各窗口订单数和订单金额，
将数据写入doris交易域省份粒度下单各窗口汇总表。

1. 数据源:
    dwd 下单表
    
2. 去重: 按照详情 id

3. 订单数的计算: 需要安装订单 id 分组, 使用状态

      详情id   sku_id    oder_id       金额   订单数
        1          1         1         100   1
        2          2         1         200   0
    
        

3. 开窗聚合
    订单金额 直接统计聚合
    
    
    
    
  
    
    
*/